// Copyright (c) Huawei Technologies Co., Ltd. 2022-2022. All rights reserved

// 服务端会话相关操作
package serversession

import (
	"fmt"
	"strconv"
	"sync"
	"time"
	"github.com/beego/beego/v2/client/orm"

	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/common"
	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/utils/log"
)

const (
	ServerSessionBackupCountToSleep = 50
	ServerSessionBackupCountToSleepSeconds = 1
)
type ServerSessionDao struct {
	sqlSession orm.Ormer
}

type ServerSessionBackupDao struct {
	sqlSession orm.Ormer
}

// NewServerSessionDao 创建一个server session dao
func NewServerSessionDao(sqlSession orm.Ormer) *ServerSessionDao {
	return &ServerSessionDao{sqlSession: sqlSession}
}

// Insert 持久化一个新的Server Session
func (s *ServerSessionDao) Insert(ss *ServerSession) (*ServerSession, error) {
	_, err := s.sqlSession.Insert(ss)
	return ss, err
}

// Delete 删除指定的server session
func (s *ServerSessionDao) Delete(ss *ServerSession) error {
	_, err := s.sqlSession.Delete(ss)
	return err
}

// Update 更新指定的server session
func (s *ServerSessionDao) Update(ss *ServerSession) (*ServerSession, error) {
	_, err := s.sqlSession.Update(ss)
	if err != nil {
		return nil, fmt.Errorf("failed to update app process for %v", err)
	}
	return ss, err
}

func (s *ServerSessionDao) UpdateStateAndReason(ss *ServerSession) (*ServerSession, error) {
	_, err := s.sqlSession.Update(ss, FieldNameState, FieldNameStateReason)
	if err != nil {
		return nil, fmt.Errorf("failed to update app process for %v", err)
	}
	return ss, err
}

// GetOneByID 根据ID查找server session
func (s *ServerSessionDao) GetOneByID(id string) (*ServerSession, error) {
	var ss ServerSession

	cond := orm.NewCondition()
	cond = cond.And(FieldNameServerSessionID, id)
	err := s.sqlSession.QueryTable(&ServerSession{}).SetCond(cond).One(&ss)
	return &ss, err
}

// ListByState1 根据指定信息查找server session列表
func (s *ServerSessionDao) ListByState1(state,
	sort string, offset, limit int) (*[]ServerSession, error) {
	var sss []ServerSession

	cond := orm.NewCondition()
	if state != "" {
		cond = cond.And("STATE", state)
	}

	_, err := s.sqlSession.QueryTable(&ServerSession{}).SetCond(cond).OrderBy(sort).
		Offset(offset).Limit(limit).All(&sss)
	return &sss, err
}

// ListByFleetIDAndInstanceIDAndProcessIDAndState 根据指定信息查找server session列表
func (s *ServerSessionDao) ListByFleetIDAndInstanceIDAndProcessIDAndState(fleetID, instanceID, processID, state,
	sort string, offset, limit int) (*[]ServerSession, error) {
	var sss []ServerSession

	cond := orm.NewCondition()
	if fleetID != "" {
		cond = cond.And("FLEET_ID", fleetID)
	}
	if instanceID != "" {
		cond = cond.And("INSTANCE_ID", instanceID)
	}
	if processID != "" {
		cond = cond.And("PROCESS_ID", processID)
	}
	// 外层查询的时候，需要把指定activating查询的时候，把creating也给返回
	switch state {
	case "":
	case common.ServerSessionStateActivating:
		cond2 := orm.NewCondition()
		cond2 = cond2.Or("STATE", common.ServerSessionStateCreating).Or("STATE", common.ServerSessionStateActivating)
		cond = cond.AndCond(cond2)
	default:
		cond = cond.And("STATE", state)
	}

	_, err := s.sqlSession.QueryTable(&ServerSession{}).SetCond(cond).OrderBy(sort).
		Offset(offset).Limit(limit).All(&sss)
	return &sss, err
}

// GetAllServerSessionCountGroupByFleetID 根据fleetID获取所有server session的计数
func (s *ServerSessionDao) GetAllServerSessionCountGroupByFleetID() (map[string]int, error) {
	serverSessionCountsByFleet := map[string]int{}

	var rows []orm.Params
	sqlStr := fmt.Sprintf("SELECT FLEET_ID, COUNT(*) as COUNT FROM %s WHERE STATE=? OR STATE=? GROUP BY FLEET_ID", TableNameServerSession)
	_, err := s.sqlSession.Raw(sqlStr, common.ServerSessionStateActivating, common.ServerSessionStateActive).Values(&rows)
	if err != nil {
		return nil, err
	}

	for i, _ := range rows {
		fleetID, _ := rows[i]["FLEET_ID"].(string)
		countStr, _ := rows[i]["COUNT"].(string)

		count, err := strconv.Atoi(countStr)
		if err != nil {
			return nil, err
		}
		serverSessionCountsByFleet[fleetID] = count
	}

	return serverSessionCountsByFleet, err
}

// QueryActivatingServerSession 查询所有处于active状态的server session
func (s *ServerSessionDao) QueryActivatingServerSession() ([]ServerSession, error) {
	var sss []ServerSession
	sqlStr := fmt.Sprintf(`select * from %s where STATE=?`, TableNameServerSession)
	log.RunLogger.Infof("sqlStr: %v", sqlStr)
	_, err := s.sqlSession.Raw(sqlStr, common.ServerSessionStateActivating).QueryRows(&sss)
	return sss, err

}

// QueryActivatingServerSessionByWorknode 查询所有处于active状态的server session
func (s *ServerSessionDao) QueryActivatingServerSessionByWorknode(worknode string) ([]ServerSession, error) {
	var sss []ServerSession
	sqlStr := fmt.Sprintf(`select * from %s where STATE=? and WORK_NODE_ID=?`, TableNameServerSession)
	log.RunLogger.Infof("sqlStr: %v", sqlStr)
	_, err := s.sqlSession.Raw(sqlStr, common.ServerSessionStateActivating, worknode).QueryRows(&sss)
	return sss, err

}

// ServerSessions ServerSession counts
func (a *ServerSessionDao) ServerSessionCounts(fleetID string) (map[string]int, error) {
	ServerSessionCounts := map[string]int{}

	var rows []orm.Params
	sqlStr := fmt.Sprintf(`SELECT %s, COUNT(*) as COUNT FROM %s WHERE %s="%s" GROUP BY %s`,
		FieldNameState, TableNameServerSession, FieldNameFleetID, fleetID, FieldNameState)
	log.RunLogger.Infof(sqlStr)
	_, err := a.sqlSession.Raw(sqlStr).Values(&rows)
	if err != nil {
		return nil, err
	}
	StateActivating := 0
	for i, _ := range rows {
		state, _ := rows[i]["STATE"].(string)
		countStr, _ := rows[i]["COUNT"].(string)

		count, err := strconv.Atoi(countStr)
		if err != nil {
			continue
		}
		if state == common.ServerSessionStateCreating || 
				state == common.ServerSessionStateActivating{
			StateActivating += count
			continue
		}
		ServerSessionCounts[state] = count
	}
	if StateActivating > 0 {
		ServerSessionCounts[common.ServerSessionStateActivating] = StateActivating
	}

	return ServerSessionCounts, err
}

// NewServerSessionBackupDao: 创建一个 server session backup dao
func NewServerSessionBackupDao(sqlSession orm.Ormer) *ServerSessionBackupDao {
	return &ServerSessionBackupDao{sqlSession: sqlSession}
}

// InsertServerSessionBackup: 把会话从运行表插入备份表中
func (b *ServerSessionBackupDao) InsertServerSessionBackup(ss *ServerSessionBackup) (*ServerSessionBackup, error) {
	_, err := b.sqlSession.Insert(ss)
	if err != nil {
		log.RunLogger.Errorf("[server session backup service] failed to insert server session %s to backup table error: %+v", 
			ss.ID, err)
		return nil, err
	}
	return ss, nil
}

// DeleteServerSessionBackup: 把会话从备份表中删除
func (b *ServerSessionBackupDao) DeleteServerSessionBackup(ss *ServerSessionBackup) error {
	_, err := b.sqlSession.Delete(ss)
	if err != nil {
		log.RunLogger.Errorf("[server session backup service] failed to delete server session %s from backup table error: %+v", 
			ss.ID, err)
		return err
	}
	return nil
}

// QueryServerSessionBackupById: 根据id从备份表中查询数据
func (b *ServerSessionBackupDao) QueryServerSessionBackupById(id string) (*ServerSessionBackup, error) {
	var ss ServerSessionBackup
	cond := orm.NewCondition()
	cond = cond.And(FieldNameServerSessionID, id)
	err := b.sqlSession.QueryTable(&ServerSessionBackup{}).SetCond(cond).One(&ss)

	return &ss, err
}

// CleanServerSessionBackup: 根据更新时间从备份表中清理dayBefore天前的数据
func (b *ServerSessionBackupDao) CleanServerSessionBackup(dayBefore int) (error) {
	var sss []ServerSessionBackup
	sqlStr := fmt.Sprintf("SELECT * FROM %s WHERE UPDATED_AT < DATE_SUB(NOW(), INTERVAL %d DAY)", 
		TableNameServerSessionBackup, dayBefore)
	toBeCleanCount, err := b.sqlSession.Raw(sqlStr).QueryRows(&sss)
	if err != nil {
		log.RunLogger.Errorf("[express data cleaner] query to be clean server session from backup table failed: %+v", err)
		return err
	}
	log.RunLogger.Infof("[express data cleaner] find %v expressed server session from backup table", toBeCleanCount)
	wg := sync.WaitGroup{}
	wg.Add(int(toBeCleanCount))
	for idx, ss := range sss {
		if idx % ServerSessionBackupCountToSleep == 0 {
			time.Sleep(ServerSessionBackupCountToSleepSeconds * time.Second)
		}
		go func(server_session ServerSessionBackup) {
			defer wg.Done()
			errD := b.DeleteServerSessionBackup(&server_session)
			if errD != nil {
				log.RunLogger.Errorf("[express data cleaner] delete server_session %s from backup table failed: %+v", 
					server_session.ID, errD)
				return
			}
			log.RunLogger.Infof("[express data cleaner] delete server_session %s from backup table success", 
				server_session.ID)
		}(ss)
	}
	wg.Wait()
	return nil
}

// BackupServerSession: 定期从运行表中备份残留的会话
func (b *ServerSessionBackupDao) BackupServerSession(dayBefore int) (error) {
	var sss []ServerSessionBackup
	sqlStr := fmt.Sprintf("SELECT * FROM %s WHERE UPDATED_AT < DATE_SUB(NOW(), INTERVAL %d DAY)", 
		TableNameServerSession, dayBefore)
	toBeBackupCount, err := b.sqlSession.Raw(sqlStr).QueryRows(&sss)
	if err != nil {
		log.RunLogger.Errorf("[express data cleaner] query to be backup server_session from runtime table failed: %+v", err)
		return err
	}
	log.RunLogger.Infof("[express data cleaner] find %v to be backup server_session from runtime table", toBeBackupCount)
	wg := sync.WaitGroup{}
	wg.Add(int(toBeBackupCount))
	for idx, ss := range sss {
		if idx % ServerSessionBackupCountToSleep == 0 {
			time.Sleep(ServerSessionBackupCountToSleepSeconds * time.Second)
		}
		go func(server_session ServerSessionBackup) {
			defer wg.Done()
			_, errI := b.InsertServerSessionBackup(&server_session)
			// 若插入失败，则可能是备份表中已存在该数据，尝试先从备份表中删除，再等待下一轮进行备份
			if errI != nil {
				log.RunLogger.Errorf("[express data cleaner] backup server session %s from runtime table to backup table failed: %+v, "+ 
					"try to delete it from backup table", server_session.ID, errI)
				backupAp, err := b.QueryServerSessionBackupById(server_session.ID)
				if err != nil {
					log.RunLogger.Errorf("[express data cleaner] query server session %s from backup table failed: %+v", 
					server_session.ID, errI)
					return
				}
				errD := b.DeleteServerSessionBackup(backupAp)
				if errD != nil {
					log.RunLogger.Errorf("[express data cleaner] delete server session %s from backup table failed: %+v", 
					backupAp.ID, errI)
					return
				}
				log.RunLogger.Infof("[express data cleaner] delete server session %s from backup table success", backupAp.ID)
				return
			}
			sqlStrA := fmt.Sprintf("DELETE FROM %s WHERE ID = '%s'", TableNameServerSession, server_session.ID)
			_, errD := b.sqlSession.Raw(sqlStrA).Exec()
			if errD != nil {
				log.RunLogger.Errorf("[express data cleaner] delete terminated server session %s from runtime table error: %+v", 
					server_session.ID, errD)
				return
			}
			log.RunLogger.Infof("[express data cleaner] backup server session %s from runtime table to backup table success", 
				server_session.ID)
		}(ss)
	}
	wg.Wait()
	return nil
}

// 从备份表中备份已被终止的进程
func (b *ServerSessionBackupDao) BackupTerminatedServerSession(Id string, wg *sync.WaitGroup) error {
	defer wg.Done()
	ss := &ServerSessionBackup{}
	sqlStr := fmt.Sprintf("SELECT * FROM %s WHERE ID='%s'", TableNameServerSession, Id)
	err := b.sqlSession.Raw(sqlStr).QueryRow(ss)
	if err != nil {
		log.RunLogger.Errorf("[express data cleaner] query terminated server session %s from runtime table error: %+v", Id, err)
		return err
	}
	// 将待清理数据插入备份表中
	_, err = b.InsertServerSessionBackup(ss)
	if err != nil {
		log.RunLogger.Errorf("[express data cleaner] insert terminated server session %s into backup table error: %+v", Id, err)
		return err
	}
	// 将待清理数据从运行表中删除
	sqlStrA := fmt.Sprintf("DELETE FROM %s WHERE ID = '%s'", TableNameServerSession, Id)
	_, err = b.sqlSession.Raw(sqlStrA).Exec()
	if err != nil {
		log.RunLogger.Errorf("[express data cleaner] delete terminated server session %s from runtime table error: %+v", Id, err)
		return err
	}
	log.RunLogger.Infof("[express data cleaner] success to backup terminated server session %s", Id)
	return nil
}